package com.jy.oms.framework.asyn.task.queue;

import com.alibaba.fastjson.JSON;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.google.common.hash.Hashing;
import com.jy.oms.framework.asyn.task.alarm.AlarmMessage;
import com.jy.oms.framework.asyn.task.alarm.AlarmMessageFacade;
import com.jy.oms.framework.asyn.task.constants.TaskStateEnum;
import com.jy.oms.framework.asyn.task.handler.ITaskHandlerManager;
import com.jy.oms.framework.asyn.task.log.ITaskLogService;
import com.jy.oms.framework.asyn.task.model.TaskDto;
import com.jy.oms.framework.asyn.task.model.TaskHandlerWrapper;
import com.jy.oms.framework.asyn.task.model.TaskResultDto;
import com.jy.oms.framework.asyn.task.model.TaskWrapper;
import com.jy.oms.framework.util.NetworkUtil;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

public abstract class AbstractTaskQueueManager implements ITaskQueueManager {
    protected static Logger logger = LoggerFactory.getLogger(AbstractTaskQueueManager.class);
    //避免告警次数过多
    private static Cache<String, Integer> alarmControlCache = Caffeine.newBuilder()
            .maximumSize(100)
            .expireAfterWrite(10, TimeUnit.MINUTES)
            .build();
    @Autowired
    protected ITaskHandlerManager taskHandlerManager;
    @Autowired
    protected ITaskLogService taskLogService;
    @Autowired
    private List<ITaskQueueFilter> taskQueueFilterList;

    @Override
    public void add(TaskWrapper taskWrapper) {
        if(CollectionUtils.isNotEmpty(taskQueueFilterList)){
            for (ITaskQueueFilter taskQueueFilter : taskQueueFilterList) {
                taskQueueFilter.filterOnAdd(taskWrapper);
            }
        }
    }

    @Override
    public void addBatch(List<TaskWrapper> taskWrapperList) {
        if(CollectionUtils.isNotEmpty(taskQueueFilterList)){
            for (ITaskQueueFilter taskQueueFilter : taskQueueFilterList) {
                taskWrapperList.forEach(taskQueueFilter::filterOnAdd);
            }
        }
    }

    protected String generateTaskIdentifier(String taskCode, String param) {
        return Hashing.murmur3_128().hashString(taskCode + "_" + param, StandardCharsets.UTF_8).toString();
    }

    protected String convertTaskParam(TaskDto taskDto) {
        String taskParam;
        if (taskDto.getParam() instanceof String) {
            taskParam = (String) taskDto.getParam();
        } else {
            taskParam = JSON.toJSONString(taskDto.getParam());
        }
        return taskParam;
    }

    protected Class<?> getTaskParamType(String taskCode) {
        TaskHandlerWrapper taskHandlerWrapper = taskHandlerManager.loadTaskHandler(taskCode);
        if (taskHandlerWrapper == null) {
            return null;
        }

        return taskHandlerWrapper.getTaskHandlerParamType();
    }

    protected Object getTaskParam(String taskCode, String param, String paramIdentifier) {
        if (StringUtils.isAnyBlank(taskCode, param)) {
            return null;
        }

        Class<?> paramType = this.getTaskParamType(taskCode);
        if (paramType == null) {
            return null;
        }

        if (paramType == String.class) {
            return param;
        }

        try {
            return JSON.parseObject(param, paramType);
        } catch (Exception e) {
            Integer num = alarmControlCache.getIfPresent(paramIdentifier);
            if (num != null) {
                alarmControlCache.put(paramIdentifier, num + 1);
                //解析报错时每5次告警一次
                if ((num + 1) % 5 == 0) {
                    this.alarmOnParseParamFailure(taskCode, paramIdentifier, e);
                }
            } else {
                alarmControlCache.put(paramIdentifier, 1);
                this.alarmOnParseParamFailure(taskCode, paramIdentifier, e);
            }
        }

        return null;
    }

    /**
     * 解析参数失败时需及时告警；
     * 对于DB队列来说，参数不合法的任务由于解析失败后不会更改状态，任务会一直被查询出来；
     * 因此DB队列任务需手动调整任务状态为已完成或失败，待修复异常参数后再调整状态为初始状态0;
     * 对于redis队列来说，需要手动从sys_task_log里取出对应的参数，修复后并手动将任务添加到redis里；
     */
    private void alarmOnParseParamFailure(String taskCode,
                                          String paramIdentifier,
                                          Exception e) {
        Integer num = alarmControlCache.getIfPresent(paramIdentifier);
        AlarmMessage alarmMessage = AlarmMessage.create()
                .title("解析参数报错")
                .add("任务taskCode", taskCode)
                .add("uuid", paramIdentifier)
                .add("报错次数", num);
        AlarmMessageFacade.getInstance().alarm(alarmMessage);
        logger.error("解析参数报错, taskCode：{}, uuid：{}", taskCode, paramIdentifier, e);
    }

    protected String getServerIp() {
        return NetworkUtil.getLocalhostStr();
    }

    protected void alarmTimeoutTask(TaskResultDto taskResultDto) {
        if (taskResultDto.getTaskState() == TaskStateEnum.TIMEOUT) {
            TaskWrapper taskWrapper = taskResultDto.getTaskWrapper();
            TaskDto taskDto = taskWrapper.getTaskDto();
            AlarmMessage alarmMessage = AlarmMessage.create()
                    .title("异步任务超时")
                    .add("任务taskCode", taskDto.getTaskCode());
            if (StringUtils.isNotBlank(taskDto.getBillCode())) {
                alarmMessage.add("任务billNo", taskDto.getBillCode());
            }
            if (StringUtils.isNotBlank(taskWrapper.getParamIdentifier())) {
                alarmMessage.add("任务uuid", taskWrapper.getParamIdentifier());
            }

            AlarmMessageFacade.getInstance().alarm(alarmMessage);
        }
    }

    private boolean isNeedInsertLog(TaskResultDto taskResultDto) {
        if (taskResultDto.isCheckFilter()) {
            return false;
        }

        //执行成功的任务添加日志暂没实际意义，根据任务状态即可判断是否执行成功
        //如果任务执行成功，回调处理报错时可能需要处理，不过也会采取另外的方案来实现
        TaskWrapper taskWrapper = taskResultDto.getTaskWrapper();
        if (taskResultDto.getTaskState() == TaskStateEnum.DONE
                && !taskWrapper.isExecuteImmediate()) {
            return false;
        }

        return true;
    }

    @Override
    public void finish(TaskResultDto taskResultDto) {
        try {
            if (this.isNeedInsertLog(taskResultDto)) {
                taskLogService.insert(taskResultDto);
            }
        } catch (Exception e) {
            logger.error("insert log error, log:{}", JSON.toJSONString(taskResultDto), e);
        }
    }

    @Override
    public void finish(List<TaskResultDto> taskResultList) {
        try {
            if (CollectionUtils.isNotEmpty(taskResultList)) {
                List<TaskResultDto> list = taskResultList.stream()
                        .filter(this::isNeedInsertLog)
                        .collect(Collectors.toList());

                if (CollectionUtils.isNotEmpty(list)) {
                    taskLogService.insert(list);
                }
            }
        } catch (Exception e) {
            logger.error("insert log error, log:{}", JSON.toJSONString(taskResultList), e);
        }
    }
}
